package com.shujia.flink.core

import java.time.Duration

import org.apache.flink.api.common.eventtime.{BoundedOutOfOrdernessWatermarks, SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object Demo3EventTime {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    //指定时间模式为事件时间
    //默认是处理时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    val lineDS: DataStream[String] = env.socketTextStream("master", 8888)


    val events: DataStream[Event] = lineDS.map(line => Event(line.split(",")(0), line.split(",")(1).toLong))

    /**
      * 统计最近5秒id的数量
      *
      */
    events
      //执行事件时间字段,水位线默认等于时间戳最大的数据的时间
      //.assignAscendingTimestamps(_.ts)
      .assignTimestampsAndWatermarks(
      //指定数据最大的延迟时间
      new BoundedOutOfOrdernessTimestampExtractor[Event](Time.seconds(5)) {
        //返回事件时间字段
        override def extractTimestamp(element: Event): Long = {
          element.ts
        }
      })
      .map(event => (event.id, 1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(5)) //5秒一个窗口
      .sum(1)
      .print()

    env.execute()


  }

  case class Event(id: String, ts: Long)

}
